/*
 * Copyright 1999-2018 Alibaba Group Holding Ltd.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.alibaba.csp.sentinel.transport.command;

import java.io.IOException;
import java.net.ServerSocket;
import java.net.Socket;
import java.net.SocketException;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import com.alibaba.csp.sentinel.command.CommandHandler;
import com.alibaba.csp.sentinel.command.CommandHandlerProvider;
import com.alibaba.csp.sentinel.concurrent.NamedThreadFactory;
import com.alibaba.csp.sentinel.transport.log.CommandCenterLog;
import com.alibaba.csp.sentinel.transport.CommandCenter;
import com.alibaba.csp.sentinel.transport.command.http.HttpEventTask;
import com.alibaba.csp.sentinel.transport.config.TransportConfig;
import com.alibaba.csp.sentinel.util.StringUtil;

/***
 * The simple command center provides service to exchange information.
 *
 * @author youji.zj
 */
public class SimpleHttpCommandCenter implements CommandCenter {

    private static final int PORT_UNINITIALIZED = -1;

    private static final int DEFAULT_SERVER_SO_TIMEOUT = 3000;
    //默认的客户端的端口是8719
    private static final int DEFAULT_PORT = 8719;

    @SuppressWarnings("rawtypes")
    private static final Map<String, CommandHandler> handlerMap = new ConcurrentHashMap<String, CommandHandler>();

    @SuppressWarnings("PMD.ThreadPoolCreationRule")
    private ExecutorService executor = Executors.newSingleThreadExecutor(
            new NamedThreadFactory("sentinel-command-center-executor"));
    private ExecutorService bizExecutor;

    private ServerSocket socketReference;

    /**
     * center在启动之前的操作，其实就是注册handdler，这个handler是什么呢？
     * 其实简单来说就是服务端发过来的请求，比如setRules,查询规则getRules等处理的handler
     *
     * @throws Exception
     */
    @Override
    @SuppressWarnings("rawtypes")
    public void beforeStart() throws Exception {
        // Register handlers
        //所以这里是注册handlers，也是通过jdk内置的spi机制进行加载的
        // CommandHandlerProvider.getInstance()得到的就是一个对象CommandHandlerProvider
        //namedHandlers()就是通过spi得到所有实现了com.alibaba.csp.sentinel.command.CommandHandler的规则处理器
        //在sentinel-transport-common的META-INF/services下的
        //com.alibaba.csp.sentinel.command.handler.BasicInfoCommandHandler
        //com.alibaba.csp.sentinel.command.handler.FetchActiveRuleCommandHandler
        //com.alibaba.csp.sentinel.command.handler.FetchClusterNodeByIdCommandHandler
        //com.alibaba.csp.sentinel.command.handler.FetchClusterNodeHumanCommandHandler
        //com.alibaba.csp.sentinel.command.handler.FetchJsonTreeCommandHandler
        //com.alibaba.csp.sentinel.command.handler.FetchOriginCommandHandler
        //com.alibaba.csp.sentinel.command.handler.FetchSimpleClusterNodeCommandHandler
        //com.alibaba.csp.sentinel.command.handler.FetchSystemStatusCommandHandler
        //com.alibaba.csp.sentinel.command.handler.FetchTreeCommandHandler
        //com.alibaba.csp.sentinel.command.handler.ModifyRulesCommandHandler
        //com.alibaba.csp.sentinel.command.handler.OnOffGetCommandHandler
        //com.alibaba.csp.sentinel.command.handler.OnOffSetCommandHandler
        //com.alibaba.csp.sentinel.command.handler.SendMetricCommandHandler
        //com.alibaba.csp.sentinel.command.handler.VersionCommandHandler
        //com.alibaba.csp.sentinel.command.handler.cluster.FetchClusterModeCommandHandler
        //com.alibaba.csp.sentinel.command.handler.cluster.ModifyClusterModeCommandHandler
        //com.alibaba.csp.sentinel.command.handler.ApiCommandHandler
        //所以这里简单来说就是通过spi找到所有实现了CommandHandler的规则处理器封装成了一个map
        //map 的key是在每个实现类上配置的注解@CommandMapping的name属性，其实就是服务端发过来请求所附带的参数比如setRules
        //map 的value就是上面的实现类的对象
        Map<String, CommandHandler> handlers = CommandHandlerProvider.getInstance().namedHandlers();
        //将找到的这个map进行注册，说的是注册，其实就是将handlers这个map放入了另外一个map 中handlerMap，key是注解@CommandMapping的name属性
        //value就是CommandHandler对应的实现类
        registerCommands(handlers);
    }

    @Override
    public void start() throws Exception {
        //得到当前服务器的cpu数
        int nThreads = Runtime.getRuntime().availableProcessors();
        //创建一个线程池
        this.bizExecutor = new ThreadPoolExecutor(nThreads, nThreads, 0L, TimeUnit.MILLISECONDS,
                new ArrayBlockingQueue<Runnable>(10),
                new NamedThreadFactory("sentinel-command-center-service-executor"),
                new RejectedExecutionHandler() {
                    @Override
                    public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
                        CommandCenterLog.info("EventTask rejected");
                        throw new RejectedExecutionException();
                    }
                });

        //创建一个算是服务端的线程Runnable，这个线程监听的是服务器dashboard发过来的请求，比如发布配置setRules
        Runnable serverInitTask = new Runnable() {
            int port;

            {
                try {
                    port = Integer.parseInt(TransportConfig.getPort());
                } catch (Exception e) {
                    port = DEFAULT_PORT;
                }
            }

            @Override
            public void run() {
                boolean success = false;
                //创建一个ServerSocket，默认端口是8719，如果被占用，依次加1，直到创建成功
                ServerSocket serverSocket = getServerSocketFromBasePort(port);

                if (serverSocket != null) {
                    CommandCenterLog.info("[CommandCenter] Begin listening at port " + serverSocket.getLocalPort());
                    socketReference = serverSocket;
                    //创建一个ServerThread线程丢到线程池中去执行
                    executor.submit(new ServerThread(serverSocket));
                    success = true;
                    port = serverSocket.getLocalPort();
                } else {
                    CommandCenterLog.info("[CommandCenter] chooses port fail, http command center will not work");
                }

                if (!success) {
                    port = PORT_UNINITIALIZED;
                }

                TransportConfig.setRuntimePort(port);
                executor.shutdown();
            }

        };

        //启动这个线程
        new Thread(serverInitTask).start();
    }

    /**
     * Get a server socket from an available port from a base port.<br>
     * Increasing on port number will occur when the port has already been used.
     *
     * @param basePort base port to start
     * @return new socket with available port
     */
    private static ServerSocket getServerSocketFromBasePort(int basePort) {
        /**
         * 这里就是通过端口来创建一个ServerSocek，这个ServerSocket是通过一定的机制来获取的
         * 简单来说就是默认的端口basePort是8719，这里设置了一个重试次数，默认是，也就是默认8719，
         * 如果8719被占用，那么在异常中加重试次数，然后重新获取ServerSocket，简单来说就是
         * 8719被占用，依次加1，直到创建ServerSocket成功
         */
        int tryCount = 0;
        while (true) {
            try {
                ServerSocket server = new ServerSocket(basePort + tryCount / 3, 100);
                server.setReuseAddress(true);
                return server;
            } catch (IOException e) {
                tryCount++;
                try {
                    TimeUnit.MILLISECONDS.sleep(30);
                } catch (InterruptedException e1) {
                    break;
                }
            }
        }
        return null;
    }

    @Override
    public void stop() throws Exception {
        if (socketReference != null) {
            try {
                socketReference.close();
            } catch (IOException e) {
                CommandCenterLog.warn("Error when releasing the server socket", e);
            }
        }
        bizExecutor.shutdownNow();
        executor.shutdownNow();
        TransportConfig.setRuntimePort(PORT_UNINITIALIZED);
        handlerMap.clear();
    }

    /**
     * Get the name set of all registered commands.
     */
    public static Set<String> getCommands() {
        return handlerMap.keySet();
    }

    class ServerThread extends Thread {

        private ServerSocket serverSocket;

        ServerThread(ServerSocket s) {
            this.serverSocket = s;
            setName("sentinel-courier-server-accept-thread");
        }

        @Override
        public void run() {
            while (true) {
                Socket socket = null;
                try {
                    //监听服务端发送过来的请求
                    socket = this.serverSocket.accept();
                    setSocketSoTimeout(socket);
                    //这里又创建了一个线程HttpEventTask，这个线程就是真正的服务器的请求的
                    HttpEventTask eventTask = new HttpEventTask(socket);
                    //创建的这个县城丢给了线程池去处理了
                    bizExecutor.submit(eventTask);
                } catch (Exception e) {
                    CommandCenterLog.info("Server error", e);
                    if (socket != null) {
                        try {
                            socket.close();
                        } catch (Exception e1) {
                            CommandCenterLog.info("Error when closing an opened socket", e1);
                        }
                    }
                    try {
                        // In case of infinite log.
                        Thread.sleep(10);
                    } catch (InterruptedException e1) {
                        // Indicates the task should stop.
                        break;
                    }
                }
            }
        }
    }

    @SuppressWarnings("rawtypes")
    public static CommandHandler getHandler(String commandName) {
        return handlerMap.get(commandName);
    }

    @SuppressWarnings("rawtypes")
    public static void registerCommands(Map<String, CommandHandler> handlerMap) {
        if (handlerMap != null) {
            for (Entry<String, CommandHandler> e : handlerMap.entrySet()) {
                registerCommand(e.getKey(), e.getValue());
            }
        }
    }

    @SuppressWarnings("rawtypes")
    public static void registerCommand(String commandName, CommandHandler handler) {
        if (StringUtil.isEmpty(commandName)) {
            return;
        }

        if (handlerMap.containsKey(commandName)) {
            CommandCenterLog.warn("Register failed (duplicate command): " + commandName);
            return;
        }

        handlerMap.put(commandName, handler);
    }

    /**
     * Avoid server thread hang, 3 seconds timeout by default.
     */
    private void setSocketSoTimeout(Socket socket) throws SocketException {
        if (socket != null) {
            socket.setSoTimeout(DEFAULT_SERVER_SO_TIMEOUT);
        }
    }
}
